package ws

import (
	"GoServer/common"
	"GoServer/message"
	"GoServer/models"
	"GoServer/utils/config"
	"GoServer/utils/databases"
	"GoServer/utils/debug"
	"GoServer/utils/tool"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"sync"
	"time"

	"github.com/gorilla/websocket"
	"gopkg.in/fatih/set.v0"
)

// /映射关系
var clientMap map[string]*models.Node = make(map[string]*models.Node, 0)

// 读写锁
var rwLocker sync.RWMutex

func init() {

}

func InitServiceWebsocket() {
	tool.Timer(
		time.Duration(config.GetConfig("app").GetInt("timeout.DelayHeartbeat"))*time.Second,
		time.Duration(config.GetConfig("app").GetInt("timeout.HeartbeatHz"))*time.Second,
		CleanConnection,
		"")

	///启动一个 Goroutine 订阅 Redis 通道
	go subscribeToChannel("live-message")
}

// 清理超时连接
func CleanConnection(param interface{}) (result bool) {
	defer func() {
		if r := recover(); !tool.InterfaceIsNil(r) {
			debug.Logln(debug.Error, "cleanConneciton err", r)
		}
	}()
	debug.Logln(debug.Info, "当前服务器在线人数：", len(clientMap))
	// 模拟并发安全性处理
	var mutex sync.Mutex
	currentTime := time.Now().Unix()
	mutex.Lock() // 上锁，防止并发访问
	for roomId := range clientMap {
		node := clientMap[roomId]
		if node.IsHeartbeatTimeOut(uint64(currentTime)) {
			debug.Logln(debug.Info, "心跳超时..... 关闭连接：", roomId)
			node.StopLiveTask(roomId, node.GameName)
			node.CloseConnect()

			// 从 clientMap 中删除超时节点
			delete(clientMap, roomId)
		}
	}
	mutex.Unlock() // 解锁
	return true
}

func OnMessage(response http.ResponseWriter, request *http.Request) {
	/// 校验token
	query := request.URL.Query()
	userId := query.Get("uid")
	roomId := query.Get("roomId")
	gameName := query.Get("gameName")
	if gameName == "" {
		gameName = common.GameName
	}

	if userId == "" {
		tool.Resp(response, message.ErrorMsg, nil, "输入 userId 错误")
		return
	}
	if roomId == "" {
		tool.Resp(response, message.ErrorMsg, nil, "输入 roomId 错误")
		return
	}
	if userId != "car_game" {
		return
	}

	bindIp := request.RemoteAddr
	//uid, err := strconv.ParseUint(bindIp, 10, 64)
	//if err != nil {
	//	fmt.Println("BindIp error :", err)
	//}
	//fmt.Println("接受到客户端消息:", bindIp, " uint64:", uid)
	debug.Logf(debug.Info, fmt.Sprintf("接受到客户端 登录消息 [userId : %s][roomId : %s][bindIp : %s]", userId, roomId, bindIp))

	isValida := true
	conn, err := (&websocket.Upgrader{
		HandshakeTimeout: 0,
		ReadBufferSize:   0,
		WriteBufferSize:  0,
		WriteBufferPool:  nil,
		Subprotocols:     nil,
		Error:            nil,
		/// token 校验
		CheckOrigin: func(r *http.Request) bool {
			return isValida
		},
		EnableCompression: false,
	}).Upgrade(response, request, nil)

	if err != nil {
		debug.Logln(debug.Error, "error : ", err)
		return
	}

	curTime := uint64(time.Now().Unix())
	/// 获取连接
	node := &models.Node{
		UserId:        userId,
		RoomId:        roomId,
		Conn:          conn,
		Addr:          conn.RemoteAddr().String(),
		HeartbeatTime: curTime,
		LoginTime:     curTime,
		DataQueue:     make(chan []byte, 50),
		GroupSets:     set.New(set.ThreadSafe),
		GameName:      gameName,
	}

	// userid 跟 node 绑定 并加锁
	rwLocker.Lock()
	clientMap[roomId] = node
	node.StartLiveTask(roomId, gameName) /// 开启直播间弹幕任务获取
	rwLocker.Unlock()

	// 完成发送的逻辑
	go sendProc(node)
	// 完成接收的逻辑
	go recvProc(node)
	// 加入在线用户到缓存列表
	//models.SetUserOnlineInfo("online_" + userId, []byte(node.Addr), time.Duration(viper.GetInt("timeout.RedisOnlineTime")) * time.Hour)
	//SendMsg(roomId, []byte("欢迎来到九叔互娱直播间"))
}

func sendProc(node *models.Node) {
	for {
		select {
		case data := <-node.DataQueue:
			debug.Logln(debug.Info, "sendProc  data :", string(data))
			err := node.Conn.WriteMessage(websocket.TextMessage, data)
			if err != nil {
				debug.Logln(debug.Error, err)
			}
		}
	}
}
func recvProc(node *models.Node) {
	for {
		_, data, err := node.Conn.ReadMessage()
		if err != nil {
			debug.Logln(debug.Info, "connetct close", err)
			rwLocker.RLock()
			delete(clientMap, node.RoomId)
			/// 停止监听任务
			rwLocker.RUnlock()
			return
		} else {
			msg := message.Message{}
			err = json.Unmarshal(data, &msg)
			debug.Logf(debug.Debug, fmt.Sprintf("[ws] <<< %s: %s", msg.RoomId, string(data)))
			if err != nil {
				debug.Logf(debug.Error, fmt.Sprintf("[ws]: [%s], [error]:[error info:%s][recv msg : %s]", node.Addr, err, string(data)))
			} else {
				debug.Logln(debug.Debug, "当前收到的消息类型：", msg.MsgType)
				if msg.MsgType == message.Heartbeat {
					currentTime := uint64(time.Now().Unix())
					node.Heartbeat(currentTime)
					debug.Logf(debug.Info, fmt.Sprintf("返回心跳数据给客户端, roomId=%s", node.RoomId))
					//SendMsg(node.RoomId, []byte("1"))
					node.DataQueue <- []byte("{'MsgType':1}")

					// 收到心跳模拟发一次礼物给客户端

				} else {
					msg.UserId = node.UserId
					msg.RoomId = node.RoomId
					bData, jsonErr := msg.MarshalBinary()
					if jsonErr != nil {
						debug.Logln(debug.Error, jsonErr)
					} else {
						WSMessageCenter(bData, node.GameName)
					}
				}
			}
		}
	}
}

func subscribeToChannel(channel string) {
	pubsub := databases.Rediss.Subscribe(context.Background(), channel)
	_, err := pubsub.Receive(context.Background())
	if err != nil {
		debug.Logln(debug.Error, "Failed to subscribe to channel:", err)
		return
	}

	// 消息接收
	ch := pubsub.Channel()

	for msg := range ch {
		fmt.Println("Received message from channel:", msg.Payload)
		// 根据需求处理消息
		data := LiveInteractiveData{}
		err := json.Unmarshal([]byte(msg.Payload), &data)
		if err != nil {
			return
		}

		debug.Logln(debug.Info, "查找当前服务器是否存在房间: ", data.RoomId)
		if IsRoomID(data.RoomId) {
			/// 这里暂时这样写， 转发弹幕消息
			if data.Message.MsgType == message.SCLiveCommentMsg {
				debug.Logln(debug.Info, "收到局域网弹幕消息：聊天消息")
				data.SendToClient(data.RoomId)
			} else if data.Message.MsgType == message.SCLiveGiftMsg {
				debug.Logln(debug.Info, "收到局域网弹幕消息：礼物消息")
				data.SendToClient(data.RoomId)
			} else if data.Message.MsgType == message.SCLiveLikeMsg {
				debug.Logln(debug.Info, "收到局域网弹幕消息：点赞消息")
				data.SendToClient(data.RoomId)
			} else if data.Message.MsgType == message.SCLiveGroupSelect {
				debug.Logln(debug.Info, "收到局域网弹幕消息：选择阵营")
				data.SendToClient(data.RoomId)
			}
		} else {
			debug.Logln(debug.Info, "没有找到此房间，不处理消息: ", data.RoomId)
		}
	}
}

// 广播消息到 Redis
func BroadcastMessage(channel string, message string) {
	err := databases.Rediss.Publish(context.Background(), channel, message).Err()
	if err != nil {
		fmt.Println("Failed to broadcast message:", err)
	}
}

func SendMsg(roomId string, msg []byte) {
	rwLocker.RLock()
	node, ok := clientMap[roomId]
	rwLocker.RUnlock()
	if ok {
		node.DataQueue <- msg
	}
}

func SendMessageToClient(roomId string, msg []byte) {
	rwLocker.RLock()
	node, ok := clientMap[roomId]
	rwLocker.RUnlock()
	if ok {
		node.DataQueue <- msg
	}
}

func GetNode(roomId string) (*models.Node, bool) {
	rwLocker.RLock()
	node, ok := clientMap[roomId]
	rwLocker.RUnlock()

	return node, ok
}

func IsRoomID(roomId string) bool {
	_, ok := clientMap[roomId]

	return ok
}
